package kafka

import (
	"context"
	"encoding/json"
	"github.com/pkg/errors"
	"github.com/segmentio/kafka-go"
	"trpg-common/config"
	"trpg-common/logger"
	"trpg-common/utils/gorm_util"
	"trpg-message/entity"
)

func StartReadingMessage() {
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:   []string{config.GetString("kafka.address")},
		Topic:     config.GetString("kafka.topic"),
		Partition: 0,
		GroupID:   "go-consumer",
		MaxBytes:  10e6, // 10MB
	})
	for {
		msg, err := reader.ReadMessage(context.Background())
		if err != nil {
			logger.Logger.Errorf("%d", errors.Wrap(err, "error in reading message from kafka"))
			break
		}
		logger.Logger.Infof("recieve message from kafka: %s", msg.Value)
		insertMessage(msg.Value)
	}
	reader.Close()
}

func insertMessage(data []byte) {
	message := entity.Message{}
	err := json.Unmarshal(data, &message)
	if err != nil {
		logger.Logger.Errorf("%+v", errors.Wrap(err, "error in converting json to message"))
	}
	_, err = gorm_util.InsertUpdateOneWithErr(&message)
}
